package com.demo.produce;

import ch.qos.logback.classic.LoggerContext;
import ch.qos.logback.classic.joran.JoranConfigurator;
import ch.qos.logback.core.util.StatusPrinter;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.LogContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.InputStream;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

/**
 * @author cxx
 * @create 2020-04-05 14:33
 **/
public class TestProduce {


    public static void main(String[] args)throws Exception {
        LoggerContext lc = (LoggerContext) LoggerFactory.getILoggerFactory();
        JoranConfigurator configurator = new JoranConfigurator();
        configurator.setContext (lc);
        lc.reset ();
        InputStream resource = TestProduce.class.getClassLoader ().getResourceAsStream ("logback.xml");
        configurator.doConfigure (resource);
//        StatusPrinter.printInCaseOfErrorsOrWarnings(lc);

        Properties prop = new Properties ();
        prop.put (ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"ikafka:9092");
        prop.put(ProducerConfig.LINGER_MS_CONFIG,"5000");

        prop.put (ProducerConfig.BATCH_SIZE_CONFIG,"200000");
        prop.put (ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
        prop.put (ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class);
        KafkaProducer producer = new KafkaProducer (prop);
        ProducerRecord<String, String> record = new ProducerRecord<String, String> ("test.sync","wwwww" );
        ProducerRecord<String, String> record2 = new ProducerRecord<String, String> ("test.sync","rrr");
        ProducerRecord<String, String> record3 = new ProducerRecord<String, String> ("test.sync","hello33");
        ProducerRecord<String, String> record4 = new ProducerRecord<String, String> ("test.sync","hello44");
        long begin = System.currentTimeMillis ();
        System.out.println (begin);
        producer.send (record,(m,e)->{if(e != null) e.printStackTrace ();});
        producer.send (record2,(m,e)->{if(e != null) e.printStackTrace ();});
        producer.send (record3,(m,e)->{if(e != null) e.printStackTrace ();});
//        producer.send (record4,(m,e)->{if(e != null) e.printStackTrace ();});
        System.out.println (System.currentTimeMillis ()-begin);
        System.out.println ("发送完成");
        TimeUnit.SECONDS.sleep (10);
        producer.close ();
        System.out.println ("===============");
    }



}
